/**
 * 
 */
package jvs.peer.xmpp;

import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import jvs.peer.io.MessageConstant;
import jvs.peer.util.Log;

import org.jivesoftware.smack.PacketCollector;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.filter.PacketFilter;
import org.jivesoftware.smack.packet.Message;
import org.jivesoftware.smack.packet.Packet;
import org.jivesoftware.smack.util.StringUtils;

/**
 * @author qiangli
 * 
 */
public class XmppPongHandler implements PacketFilter {
	public class Selector {
		private CountDownLatch latch = new CountDownLatch(1);
		private long time = System.currentTimeMillis();
		private boolean alive = false;
		private String peer;

		private Selector(String peer) {
			this.peer = peer;
		}

		public boolean select() throws InterruptedException {
			latch.await();
			return alive;
		}

		public boolean select(long time) throws InterruptedException {
			latch.await(time, TimeUnit.SECONDS);
			return alive;
		}

		@Override
		public String toString() {
			return "peer: " + peer + " alive: " + alive + " time: "
					+ new java.util.Date(time);
		}

	}

	private class Worker extends Thread {
		public void run() {
			while (running) {
				try {
					Packet pkt = collector.nextResult();
					String from = pkt.getFrom();
					String peer = StringUtils.parseBareAddress(from);
					if (Log.DEBUG) {
						Log.log(this, "@@@  from: "+ from + " peer: " + peer
							+ " id: " + pkt.getPacketID());
					}
					Selector sel = selectors.get(peer);
					if (sel != null) {
						sel.alive = true;
						sel.latch.countDown();
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}

	private class Cleaner extends Thread {
		public void run() {
			while (running) {
				try {
					long now = System.currentTimeMillis();
					for (Iterator<Map.Entry<String, Selector>> it = selectors.entrySet().iterator(); it.hasNext(); ){
						Map.Entry<String, Selector> e = it.next();
						Selector sel = e.getValue();
						long elapsed = now - sel.time;
						if (elapsed > maxElapsed) {
							sel.latch.countDown();
							it.remove();
							if (Log.DEBUG) {
								Log.log(this, " removed " + sel);
							}
						}
					}
					Thread.sleep(maxElapsed);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}

	private final static String MSG = "ping";

	private Map<String, Selector> selectors = Collections
			.synchronizedMap(new HashMap<String, Selector>());

	private volatile boolean running;

	private XMPPConnection conn;

	private PacketCollector collector;

	private long maxElapsed = 3000; // e seconds

	private static long instance = 0;

	private static String prefix = StringUtils.randomString(5);

	private static synchronized String nextId() {
		return "ping-" + prefix + Long.toString(instance++);
	}

	public XmppPongHandler(XMPPConnection conn) {
		this.conn = conn;
	}

	/**
	 * Only accept pong packet
	 */
	public boolean accept(Packet packet) {
		//allow loopback
		Object obj = packet.getProperty(MessageConstant.PONG);
		String from = packet.getFrom();
		String id = packet.getPacketID();
		if (Log.DEBUG) {
			Log.log(this, " accepting id: " + id + " from: "
				+ from + " to: " + packet.getTo());
		}
		return (obj != null && from != null && id != null);
	}

	/**
	 * 
	 * @param id
	 * @param peer
	 * @throws InterruptedException
	 */
	synchronized public Selector ping(String from) {
		String peer = StringUtils.parseBareAddress(from);
		Selector sel = selectors.get(peer);
		if (sel == null) {
			Message msg = new Message(peer);
			msg.setThread(nextId());
			msg.setProperty(MessageConstant.PING, MSG);
			//
			msg.setSubject(MSG);
			msg.setBody(MSG);
			msg.setType(Message.Type.headline);
			//
			sel = new Selector(peer);
			selectors.put(peer, sel);
			//
			conn.sendPacket(msg);
		}
		return sel;
	}

	private Thread worker;
	private Thread cleaner;

	public void startup() {
		running = true;
		collector = conn.createPacketCollector(this);
		//
		worker = new Worker();
		cleaner = new Cleaner();
		worker.start();
		cleaner.start();
	}

	public void shutdown() {
		running = false;
		collector.cancel();
		worker.interrupt();
		cleaner.interrupt();
		conn = null;
	}
}
